package org.budo.dubbo.protocol.async.repository.aliyun.mns;

import java.util.List;

import org.budo.dubbo.protocol.async.repository.AsyncRepository;
import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.support.lang.util.StringUtil;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.aliyun.mns.client.AsyncResult;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.model.Message;

/**
 * @author lmw
 */
public class AliyunMnsAsyncRepository extends AbstractAliyunMnsAsyncRepository {
    private static final Logger log = LoggerFactory.getLogger(AliyunMnsAsyncRepository.class);

    /**
     * @see com.aliyun.mns.client.CloudQueue#batchPopMessage(int)
     * @see com.aliyun.mns.client.CloudQueue#batchPopMessage(int, int)
     */
    @Override
    public Integer popMessageAndInvoke(Invoker invoker) {
        Class<?> interfaceType = invoker.getInterface();
        String destinationName = invoker.getUrl().getParameter(AsyncRepository.DESTINATION_NAME);
        if (StringUtil.isEmpty(destinationName)) {
            destinationName = "budo-async-" + interfaceType.getName();
        }

        CloudQueue cloudQueue = this.getInterfaceQueue(interfaceType, destinationName); // 每个接口一个频道

        Integer popBatchSize = this.getPopBatchSize();
        Integer popWaitSeconds = this.getPopWaitSeconds();

        List<Message> messages = cloudQueue.batchPopMessage(popBatchSize, popWaitSeconds);

        PopCallback popCallback = new PopCallback(cloudQueue, invoker, this);
        popCallback.onSuccess(messages); // 循环处理消息

        return null == messages ? 0 : messages.size();
    }

    @Override
    public void sendMessageWhenInvoke(Invocation invocation) {
        Class<?> interfaceType = invocation.getInvoker().getInterface();
        String destinationName = invocation.getInvoker().getUrl().getParameter(AsyncRepository.DESTINATION_NAME);
        if (StringUtil.isEmpty(destinationName)) {
            destinationName = "budo-async-" + interfaceType.getName();
        }

        CloudQueue cloudQueue = this.getInterfaceQueue(interfaceType, destinationName); // 每个接口一个频道

        BudoAsyncInvocation budoAsyncInvocation = BudoAsyncInvocation.fromDubboInvocation(invocation);
        byte[] requestBody = this.serializeAsyncInvocation(budoAsyncInvocation, invocation.getInvoker());
        Message message = new Message(requestBody);

        // 发消息
        AsyncResult<Message> asyncResult = cloudQueue.asyncPutMessage(message, new PutCallback(message));

        if (log.isDebugEnabled()) {
            log.debug("#55 asyncResult=" + asyncResult.getClass().getName() + "@" + Integer.toHexString(asyncResult.hashCode()));
        }
    }
}